package a_aa_amainpackage.lvyou_clear.lv_update

import com.alibaba.fastjson.JSON
import com.o2o.utils.Iargs
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/6/5 16:23
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object CheckTuniuESData {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .config("spark.debug.maxToStringFields", "2000")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.caseSensitive", "true")
      .config("es.nodes", "192.168.1.29")
      .config("es.port", "9200")
      .config("cluster.name","O2OElastic")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
      .master("local[*]")
      .getOrCreate()

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", Iargs.OBSACCESS)
    sc.hadoopConfiguration.set("fs.s3a.secret.key", Iargs.OBSSECRET)
    sc.hadoopConfiguration.set("fs.s3a.endpoint", Iargs.OBSENDPOINT)
    sc.setLogLevel("WARN")
    var year ="2021"
    var month = "3"

    import org.elasticsearch.spark._


    val data1 = sc.esJsonRDD(s"${year}_update_tuniu_ly/tuniu_ly_${year}_${month}",
      """
        |{
        |  "query": {
        |    "bool": {
        |      "must_not": [
        |        {
        |          "bool": {
        |            "filter": {
        |              "exists": {
        |                "field": "province_out"
        |              }
        |            }
        |          }
        |        }
        |      ]
        |    }
        |  }
        |}
      """.stripMargin).values

    val frame: DataFrame = util.updaue_addressfirst.addressclear(sc,spark.sqlContext,spark.sqlContext.sparkSession.read.json(data1),"ly")


    val value3 = frame.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)

//      lines.put("Country_out","中国")

      lines
    })
    value3.saveToEs(s"${year}_update_tuniu_ly/tuniu_ly_${year}_${month}", Map("es.mapping.id" -> "product_Id"))


    /**
      * 查询国外Country_out为0的，first_arriving第一出发地为中国的，通过 util.updaue_addressfirst.addressclear进行匹配
      * {
  "query": {
    "bool": {
      "must": [
        {"match_phrase": {
          "toursim_tickets_type": "03002"
        }},
        {"match_phrase": {
          "Country_out": "0"
        }}
      ]
    }
  }
}
      */



    /**
      * 查询中国的，国家字段为0，把Country_out字段打上中国即可
   {
  "query": {
    "bool": {
      "must": [
        {"match_phrase": {
          "toursim_tickets_type": "03001"
        }},
        {"match_phrase": {
          "Country_out": "0"
        }}
      ]
    }
  }
}
      */


    /**    这种情况是没有国家、省份、城市字段，需要提取first_arriving给泽民，他做好映射之后，在通过 util.updaue_addressfirst.addressclear进行匹配，记得注释掉
   {
  "query": {
    "bool": {
      "must_not": [
        {
          "bool": {
            "filter": {
              "exists": {
                "field": "province_out"
              }
            }
          }
        }
      ]
    }
  }
}
      */

    sc.stop()
}
}
